#include "dbenv.hxx"
#include "db/dbpartition.hxx"
#include "db/datastore_impl.hxx"
#include "log/logmanager.hxx"
#include "db/datasegmentmanager.hxx"
#include "db/staticindexarray.hxx"
#include "db/addressformat.hxx"
#include "db/segmentdataparser.hxx"
#include "db/hash.hxx"
#include "db/datasegment.hxx"
#include "log/nodatalogmanager.hxx"
#include "log/logfile.hxx"
#include "util/lockguard.hxx"
#include "db/compact_impl.hxx"
#include "util/common.hxx"
#include "util/logger.hxx"
#include <iostream>
#include <algorithm>
#include <errno.h>
#include <assert.h>
#include <memory>


DbPartition::DbPartition(DataStoreImpl& ds, 
                         LogManager& logMgm,  
                         const std::string& partitionName)
: m_partitionName(partitionName), 
  m_ds(&ds), 
  m_logmgr(&logMgm), 
  m_segmentMgr(NULL), 
  m_indexarray(NULL) {

    m_good = 0; 
	std::string dsPath = ds.getDsPath();
	m_partitionPath = dsPath + "/" + partitionName;
}


DbPartition::~DbPartition() {

	if(m_segmentMgr) {delete m_segmentMgr; m_segmentMgr = NULL;}
	if(m_indexarray) {delete m_indexarray; m_indexarray = NULL;}

    pthread_mutex_destroy(&m_mutex);
    
    
    std::cout<<m_good<<"\n";
}


DbStatus DbPartition::init() {

    int rc = pthread_mutex_init(&m_mutex, NULL);
    if(rc != 0) return DbStatus("DbPartition::init::pthread_mutex_init", rc);

    uint32_t flag = m_ds->getFlag();
    if(flag & INDEXARRAY_STATIC) {
	    m_indexarray = new StaticIndexArray(m_partitionPath, m_ds->getIndexArrayCapacity());
    }
    else {
        return DbStatus("DbPartition::init", DbStatus::DB_NOTSUPPORTED);
    }

	DbStatus ret = m_indexarray->init();
	if(!ret.isok()) return ret;
		
	m_segmentMgr = new DataSegmentManager(*m_ds, *this);

	return m_segmentMgr->init();
}

DbStatus DbPartition::flush() {

	DbStatus ret;
    
    //get list first before index flush
    std::set<uint16_t> s = m_segmentMgr->getSegmentIDsNotAlive();
    
    ret = m_segmentMgr->flush(); if(!ret.isok()) return ret;
	
    ret = m_indexarray->flush(); if(!ret.isok()) return ret;
 
    for(std::set<uint16_t>::iterator iter = s.begin(); iter != s.end(); 
            ++iter) {

        ret =  m_segmentMgr->discardSegment(*iter); if(!ret.isok()) return ret;
    }        

    return ret;
}


DbStatus DbPartition::flushSegments() {

    return m_segmentMgr->flush();
}

DbStatus DbPartition::flushIndexArray() {

    DbStatus ret;

    //get list first before index flush
    std::set<uint16_t> s = m_segmentMgr->getSegmentIDsNotAlive();

    ret = m_indexarray->flush();

    for(std::set<uint16_t>::iterator iter = s.begin(); iter != s.end();
            ++iter) {

        ret =  m_segmentMgr->discardSegment(*iter); if(!ret.isok()) return ret;
    }

    return ret;
}


DbStatus DbPartition::get(Dbt& key, Dbt& value) {

    DbStatus ret;
    
    uint32_t index = getIndex(key);
	uint64_t addr;
    ret = getAddr(index, addr);
    if(!ret.isok()) return ret;
	if(addr == 0) return DbStatus("DbPartition::get", DbStatus::DB_KEYNOTFOUND);
	
	AddressFormat af(addr);

	uint16_t segID = af.getSegID();
	uint32_t offset = af.getOffset();
	uint32_t dataSize = af.getDataSize();


	//no lock for the get() operation. It is possible that the segment is
    //deleted by compact between getAddr and reading from segment.
	//TODO: we should retry one more time if get() report a SEGNMENTNOT FOUND error
	Dbt result;
    ret = m_segmentMgr->get(segID, offset, dataSize, result);
	if(!ret.isok()) return ret;

    //check buffer size of the value dbt
    if(value.empty() == false && value.getSize() < dataSize) return
        DbStatus("DbPartition::get", DbStatus::DB_BUFFERWILLOVERFLOW);

    //reuse the buffer
    if(value.empty()) {
        value = result; value.setDatalen(0);
    }
	
	//extract the value matched with the key
	bool succ = SegmentDataParser::getValueByKeyDbt(key, result, value);
	if(!succ) return DbStatus("DbPartition::get", DbStatus::DB_KEYNOTFOUND);


	return ret;
}


DbStatus DbPartition::putInternal(Dbt& key, Dbt& value, bool deletion) {

    uint64_t wt = CompactImpl::getInstance()->timeToWait();
    if(wt > 0) utility::sleepnano(wt);
    
    uint32_t index = getIndex(key);

    uint16_t newSegID, oldSegID = 0; 
    uint32_t loadSizeOld = -1, loadSizeNew;
    Dbt oldData, newData;
    uint64_t oldAddr, newAddr;
    uint32_t oldDataSize, newDataSize;
    uint32_t oldOffset, newOffset;
    bool needPut = true;
    utility::auto_array_ptr<char> memOld;

    DbStatus ret;

    LockGuard lockg(m_mutex);
    
    ret = getAddr(index, oldAddr);
    if(!ret.isok()) return ret; 

    if(oldAddr) {

        //at most try 3 times
        int MAXTRY = 3;

        int i = 0;
        for(; i < MAXTRY; ++i){
            
            needPut = true;

            //in order to improve concurrency, release lock because will do IO
            lockg.reset();

            AddressFormat af(oldAddr);

            oldSegID = af.getSegID();
            oldOffset = af.getOffset();
            oldDataSize = af.getDataSize();

            //allocate enough buffer for the old data and new data, in order to
            //reduce the number of memory allocation
            if(oldDataSize > 0) {
                
                uint32_t bufSize = oldDataSize;
                if(!deletion) bufSize += sizeof(uint32_t)*2 + key.getDatalen() + value.getDatalen(); 
                char* buf = new char[bufSize]; 
                memOld.reset(buf);
                oldData.setBuffer(buf, bufSize, 0, DBT_MALLOC);
            }

            ret = m_segmentMgr->get(oldSegID, oldOffset, oldDataSize, oldData);  if(!ret.isok()) return ret;
            if(deletion) {

                newData = oldData;
                SegmentDataParser::removeByKeyDbt(key, newData);
                if(oldData.getDatalen() == newData.getDatalen()) return DbStatus("del", DbStatus::DB_KEYNOTFOUND);
                if(newData.getDatalen() == 0) {
                    needPut = false;
                    newAddr = 0;
                }
            }
            else {
                newData = SegmentDataParser::serializeDbtFromDbt(key, value, oldData);
            }

            //acquire lock again
            lockg.set(m_mutex);

            uint64_t addr2;
            ret = getAddr(index, addr2);
            if(!ret.isok()) return ret;
            if(addr2 == oldAddr)  break; 
            oldAddr = addr2;

        }


        if(i == MAXTRY) return DbStatus("DbPartition:: put", EAGAIN);
    }
    else {

        m_good++;
        if(deletion) return DbStatus("del", DbStatus::DB_KEYNOTFOUND);
        newData = SegmentDataParser::serializeDbt(key, value, newData);
    }


    //the buffer of newData will be released after appending LogRecord. If new
    //buffer is different from old buffer, we need delete old
    //buffer. But if they are same, we should reset memOld
    if(newData.data() == oldData.data())
        memOld.release();

    if(needPut) {
        //TODO: we may need to release lock here
        ret = m_segmentMgr->put(newData, newSegID, newOffset, loadSizeNew);
        if(!ret.isok()) return ret;

        assert(newSegID > 0);
        AddressFormat af(newSegID, newOffset, newData.getDatalen());
        newAddr = af.getAddr();

        //   std::cout<<"put new seg "<<newSegID<<" "<<newOffset<<" "<<newData.getDatalen()<<"\n";
     //   std::cout<<"put old seg "<<oldSegID<<" "<<oldOffset<<" "<<oldDataSize<<"\n";
    }


    //decrease load size in orginal old segment, which is in buffer
    //update the data size of the old(source) segment after put succeed
    if(oldAddr) {
      
        DataSegment* seg = m_segmentMgr->getSegment(oldSegID);
        loadSizeOld = seg->changeDataSize(0 - oldDataSize - sizeof(uint32_t)); 

        if(oldSegID != newSegID && CompactImpl::getInstance()->isEnabled() ) {
       
            seg->notifyCompact();
        }
    }

    //append log, get scn
    LogRecord rec;
    rec.setType(LogRecord::PUT);
    rec.setDsName(m_ds->getDsName());
    rec.setPartitionName(m_partitionName);
    rec.setLoadSizeOld(loadSizeOld);
    rec.setLoadSizeNew(loadSizeNew);
   
    Dbt dbtnewdata = newData;
    if(m_logmgr->getFlag() & LOG_NO_DATA ) { 
    
        dbtnewdata.setDatalen(0);
    }
    
    rec.setDataNew(dbtnewdata);
    

    rec.setIndex(index);
    rec.setAddrNew(newAddr);
    rec.setAddrOld(oldAddr);
    
    ret = m_logmgr->addRecord(rec);
    if(!ret.isok()) {

        //writing log should not fail. 
        //TODO: If it fails, we should restart
        return ret;
    }

    uint64_t scn = rec.getScn();
    rec.clear();

    //update index array, entry and hwm
    ret = m_indexarray->set(index, newAddr, scn); 

    return ret;
}

DbStatus DbPartition::put(Dbt& key, Dbt& value) {
    return putInternal(key, value, false);
}

DbStatus DbPartition::del(Dbt& key) {
    
    Dbt v;
    return putInternal(key, v, true);
}



uint32_t DbPartition::getIndex(Dbt& key) {

	Fnv1Hash32 h;
    //MurmurHash h; 

	uint32_t hashCode = h.hash(key);
	uint32_t index = hashCode%m_indexarray->getCapacity();

	return index;
}

DbStatus DbPartition::getAddr(uint32_t index, uint64_t& addr) {

	return m_indexarray->get(index, addr);
}


void DbPartition::getWaterMarks(uint64_t& lwm, uint64_t& hwm) {

    m_indexarray->getWaterMarks(lwm, hwm);
}

//for recovery only
DbStatus DbPartition::get(uint16_t segID,
                          uint32_t offset, 
                          uint32_t dataSize,
                          std::string& data) {

    DbStatus ret;

    ret = m_segmentMgr->get(segID, offset, dataSize, data);

    return ret;     
}


DbStatus DbPartition::put(uint16_t segID, 
                          uint32_t offset, 
                          const std::string& data, 
                          uint32_t loadSize, 
                          uint32_t index, 
                          uint64_t addrvalue, 
                          uint64_t scn) {

	DbStatus ret;

	DataSegment* seg = m_segmentMgr->getSegment(segID);

	if(seg == NULL) {
	
        Logger::log("cannot find segment: %d", segID);
		return DbStatus("DbPartition::put", DbStatus::DB_SEGMENTNOTFOUND);
	}

	//data
    if( !(m_logmgr->getFlag() & LOG_NO_DATA) ) {
        
        ret = seg->write(offset, data);
        if(!ret.isok()) return ret;
    }

	//loadsize
	seg->setDataSize(loadSize);

	//indexarray
	ret = m_indexarray->set(index, addrvalue, scn);

	return ret;
}


DbStatus DbPartition::setLoadSize(uint16_t segID, uint32_t loadSize) {

    DbStatus ret;

	DataSegment* seg = m_segmentMgr->getSegment(segID);

	if(seg == NULL) {
	
        Logger::log("cannot find segment: %d", segID);
		return DbStatus("DbPartition::put", DbStatus::DB_SEGMENTNOTFOUND);
	}
	
	seg->setDataSize(loadSize);
    
    return ret;
}

std::map<uint16_t, std::vector<CompactIndexItem> >
DbPartition::getCompactIndexItemMap(std::set<uint16_t>& segIDList,
        CompactOption& option) {

    std::map<uint16_t, std::vector<CompactIndexItem> > table;

    uint32_t indexCapacity = m_indexarray->getCapacity();
    for(int i = 0; i < indexCapacity && !option.shouldDone(); ++i) {
        
        uint64_t addr;
        getAddr(i, addr);

        AddressFormat af(addr);
        uint16_t segID = af.getSegID();
        if(segIDList.count(segID) == 0) continue;

        if(table.count(segID) == 0) table[segID] = std::vector<CompactIndexItem>();

        //no lock is used here, addr may be changed by put/del so addr may not
        //matched with segID later before compactInternal
        CompactIndexItem item(i, addr);
        table[segID].push_back(item);
    }

    return table;
}

DbStatus DbPartition::compact(CompactOption& option) {
    
    DbStatus ret;

    std::set<uint16_t> segIDList = m_segmentMgr->getSegmentIDsForCompact(option.m_validDataRatio);

    std::map<uint16_t, std::vector<CompactIndexItem> > table =
        getCompactIndexItemMap(segIDList, option);


    if(table.empty() || option.shouldDone()) return ret;

    option.m_compacting = true;
    option.m_segs = table.size();

    for(std::map<uint16_t, std::vector<CompactIndexItem> >::iterator iter =
            table.begin(); iter != table.end() && !option.shouldDone(); ++iter) {
       
        ret = compactInternal(option, iter->first, iter->second);
        if(!ret.isok()) break;
        
        option.m_segs--;
        Logger::log("segment %d is compacted", iter->first);
    }

    option.m_segs = 0;
    option.m_compacting = false;

    //verify
    //std::vector<uint16_t> v;

    for(std::set<uint16_t>::iterator iter = segIDList.begin(); iter !=
            segIDList.end(); ++iter) {

        DataSegment* seg = m_segmentMgr->getSegment(*iter);
        if(seg && seg->getActualDataSize() == 0) {
            //change the field 'live' to 0
            seg->setAlive(false);
            seg->dropCache();
            Logger::log("segment %d is set to not-alive", *iter);
            //v.push_back(*iter);
        }
    }

/*
    //sync when finish compacting a partition
    if(m_logmgr->getFlag() & LOG_NO_DATA ) {
    
//        (dynamic_cast<NodataLogManager*>(m_logmgr))->invokeSync();    
    }
    else {
        
        switchEntries();
        m_logmgr->flush();
        flush();
    }

    //delete the compacted segment to reclaim space, 
    for(int i = 0; i < v.size(); ++i) {
        Logger::log("compact remove the segID: %d", v[i]);
        m_segmentMgr->discardSegment(v[i]);
    }
*/
    return ret;
}

DbStatus DbPartition::compactInternal(CompactOption& option, uint16_t segID,
        std::vector<CompactIndexItem>& vCompactItem) {
   
    DbStatus ret;

    std::sort(vCompactItem.begin(), vCompactItem.end(), CompactIndexCompartor());

    DataSegment* seg = m_segmentMgr->getSegment(segID);

    for(int i = 0; i < vCompactItem.size() && !option.shouldDone(); ++i) {

        uint64_t wtime = 1;
        while(wtime > 0) {
            wtime = option.m_throttling.getToken();
            if(wtime) utility::sleepnano(wtime*1000);
        }

        ret = putForCompact(seg, vCompactItem[i]);

        if(!ret.isok()) return ret;
    }

    return ret;
}

//right now, copmact need to compete with put/del operations, which will degrade
//the throughput of put/del operations. To improve it, first we can collect the data
//needed to be compacted and insert each data compaction into a job
//bactch (the two actions don't need lock). Then, each put/del operation steal a
//bactch from compaction. The stealing approach can reduce context switch.
DbStatus DbPartition::putForCompact(DataSegment* seg, CompactIndexItem& item) {
    
    DbStatus ret;

    uint32_t index = item.m_index;

    uint16_t newSegID, oldSegID = seg->getSegId();
    uint32_t loadSizeOld = -1, loadSizeNew;
    std::string oldData, newData;
    uint64_t oldAddr, newAddr;
    uint32_t oldDataSize, newDataSize;
    uint32_t oldOffset, newOffset;

   
    oldAddr = item.m_addr; 

    uint64_t curAddr;
    ret = getAddr(index, curAddr);
    if(!ret.isok()) return ret;

    //index addr value was changed, so don't need move the offset any more
    if(curAddr != oldAddr) return ret;

    LockGuard lockg;
    
    AddressFormat af(oldAddr);
    oldOffset = af.getOffset();
    oldDataSize = af.getDataSize();

    ret = m_segmentMgr->get(oldSegID, oldOffset, oldDataSize, oldData);  if(!ret.isok()) return ret; 
    newData = oldData;

    ret = m_segmentMgr->put(newData, newSegID, newOffset, loadSizeNew);
    if(!ret.isok()) return ret;

    AddressFormat naf(newSegID, newOffset, newData.size());
    newAddr = naf.getAddr();

    //acquire lock
    lockg.set(m_mutex);

    uint64_t addr2;
    ret = getAddr(index, addr2);
    if(!ret.isok()) return ret;

    if(addr2 != oldAddr) {

        //should reduce data size of the active compact segment
        DataSegment* compactSeg = m_segmentMgr->getSegment(newSegID);
        uint32_t payloadLen = sizeof(uint32_t) + newData.size();
        compactSeg->changeDataSize(0 - payloadLen); 

        return ret;
    }

    loadSizeOld = seg->changeDataSize(0 - oldDataSize - sizeof(uint32_t));

    //compact doesn't need to write log for LOG_NO_DATA, because segment is
    //flushed before log flush
    if( !(m_logmgr->getFlag() & LOG_NO_DATA) ) {
        //append log, get scn
        LogRecord rec;
        rec.setType(LogRecord::COMPACT);
        rec.setDsName(m_ds->getDsName());
        rec.setPartitionName(m_partitionName);
        rec.setLoadSizeOld(loadSizeOld);
        rec.setLoadSizeNew(loadSizeNew);

        Dbt dbtnewdata = newData;
        dbtnewdata.setDatalen(0);

        rec.setDataNew(dbtnewdata);


        rec.setIndex(index);
        rec.setAddrNew(newAddr);
        rec.setAddrOld(oldAddr);

        ret = m_logmgr->addRecord(rec);
        if(!ret.isok()) {

            //writing log should not fail. 
            //TODO: If it fails, we should restart
            return ret;
        }

        uint64_t scn = rec.getScn();
        rec.clear();
        ret = m_indexarray->set(index, newAddr, scn);
    }
    else {

        //update index array, entry and but don't change hwm
        ret = m_indexarray->set(index, newAddr);
    }

//        std::cout<<"compact new seg "<<newSegID<<" "<<newOffset<<" "<<newData.size()<<"\n";
//        std::cout<<"compact old seg "<<oldSegID<<" "<<oldOffset<<" "<<loadSizeOld<<"\n"; //<<oldDataSize<<"\n";

    return ret;
}

//protected by env sync lock
void DbPartition::switchEntries() {

    m_indexarray->switchEntries();
}


